package org.example.day20241227.operators;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.impl.StreamFormatAdapter;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.File;

/**
 * @className: FilterDemo
 * @author: 孙磊
 * @date: 2024/12/29 22:14
 * @Version: 1.0
 * @description: 为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素。
 * 1.3.filter算子,读一个数组数据,过滤出大于10或过滤一些特殊字符的数据,得到一个新的数据集;
 */
public class FilterDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        BulkFormat<String, FileSourceSplit> bulkFormat =
                new StreamFormatAdapter<>(new TextLineInputFormat());

        FileSource<String> source = FileSource
                .forBulkFileFormat(bulkFormat, new Path("src/main/resources/2024-12-29--22/part-5f90868c-fe15-4ddf-99c6-3c37e7385d2e-0"))
                .build();

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "filter-source")
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        //获取字符串长度超过3的单词，不超过3的认为是无效单词
                        return s.length() > 3;
                    }
                }).print();
        env.execute();
    }
}
